package net.bwie.realtime.jtp.dws.log.fuction;

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class PageViewWindowFunction
        implements WindowFunction<PageViewBean,String,String, TimeWindow> {
    private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    @Override
    public void apply(String s, TimeWindow timeWindow,
                      Iterable<PageViewBean> iterable, Collector<String> collector) throws Exception {
        // 1.获取窗口开始时间和结束时间
        String windowStartTime = format.format(timeWindow.getStart());
        String windowEndTime = format.format(timeWindow.getEnd());
        // 2.对窗口中数据计算
        Long pvCount = 0L;
        Long pvDuringTime = 0L;
        Long uvCount = 0L;
        Long sessionCount = 0L;
        for(PageViewBean pageViewBean : iterable){
            pvCount += pageViewBean.getPvCount();
            pvDuringTime += pageViewBean.getPvDuration();
            uvCount += pageViewBean.getUvCount();
            sessionCount += pageViewBean.getSessionCount();
        }
        // 3.输出
        String output = windowStartTime + "," + windowEndTime+","
                +windowStartTime.substring(0,10) + ","
                + s + ","
                + uvCount + "," +sessionCount + "," + pvCount + "," +pvDuringTime;
        collector.collect(output);
    }
}
